package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.DorisUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2023/9/1
 * sku粒度下单聚合统计
 * 维度：sku
 * 度量：原始金额、优惠券减免金额、活动减免金额、实付金额
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、Doris、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发流程
 *      基本环境准备
 *      检查点相关设置
 *      从kafka主题中读取数据
 *      对读取的数据进行类型转换并过滤掉空消息
 *      按照唯一键(订单明细的id)进行分组
 *      去重
 *          为什么会产生重复数据？
 *              因为我们数据来源是dwd_trade_order_detail,下单事实表是由订单明细、订单、明细活动、明细优惠券
 *              4张表组成的，这4张表在进行关联的时候用到了左外连接，如果左表数据先到，右表数据后到，会产生以下3条数据
 *                  左表      null    +I
 *                  左表      null    -D
 *                  左表      右表     +I
 *              这样的数据写到kafka主题中，主题会接收到3条消息
 *                  左表      null
 *                  null
 *                  左表      右表
 *              所以我们在从这样的主题读取数据的时候，需要过滤空以及去重
 *          去重方式1：状态 + 定时器
 *              当第一条数据到来的时候，先不向下游传递，先将其放到状态中，并注册一个5s后执行的定时器，当有重复
 *              数据到来的时候，会用状态中数据的聚合时间和这条数据的聚合时间进行比较，将时间大的放到状态中，
 *              等到定时器被触发的时候，将状态的数据写到下游
 *              不足：每条数据到来的时候，都要等5s，时效性较差
 *          去重方式2：状态 + 抵消
 *              数据到来的时候，直接发送到下游，并放到状态中，如果有重复数据的话，将状态中已经发送到下游的数据影响到
 *              度量值的属性进行取反，再发送到下游。
 *              同时第二条重复数据发送到下游
 *              优点：时效性强
 *              缺点：如果数据有重复的话，发送3条数据到下游
 *      对流中数据进行类型转换      jsonObj->统计的实体类对象
 *      执行Watermark以及提取事件时间字段
 *      分组
 *      开窗
 *      聚合计算
 *      维度关联
 *          基本的维度关联的实现
 *              HbaseUtil --- JSONObject getDimInfoFromHbase()
 *          优化1：旁路缓存
 *              思路：先从缓存中查询维度数据，如果缓存中找到了要关联的维度数据，直接将其作为方法的返回值进行返回；
 *              如果在缓存中没有找到要关联的维度，发送请求到hbase中获取维度数据，并将获取到维度数据放到缓存进行
 *              缓存，方便下次查询使用。
 *              选型：
 *                  状态         性能很好，维护性差
 *                  redis       性能不错，维护性好  √
 *              关于Redis的配置
 *                  key： 维度表名:主键值
 *                  type：String
 *                  ex：  1day
 *                  注意：如果维度数据发生了变化，将Redis中缓存的数据清除掉
 *          优化2：异步IO
 *              为什么使用异步：
 *                  在使用map算子对流中数据进行处理的时候，在单个并行度上使用的是同步的方式，处理完一个元素之后，再
 *              处理下一个元素，等待的时间较长，可以发送异步请求，同时对流中元素进行处理，这样等待的时间会被其它
 *              元素发送请求的时间稀释，提升关联效率；如果加大map算子的并行度，也可以提升处理能力，但是更多并行度，
 *              意味着更多的硬件资源，不可能无限制的提升。所以我们通过异步的方式提升处理能力
 *              Flink中如何发送异步请求
 *                  AsyncDataStream.[un]orderedWait(
 *                      流,
 *                      如何发送异步请求 implements AsyncFunction---asyncInvoke,
 *                      超时时间，
 *                      时间单位
 *                  )
 *              RedisUtil
 *                  获取支持异步操作的连接对象
 *                  关闭支持异步操作的连接对象
 *                  通过异步的方式从Redis中读取数据
 *                  通过异步的方式向Redis中写数据
 *              HbaseUtil
 *                  获取支持异步操作的连接对象
 *                  关闭支持异步操作的连接对象
 *                  通过异步的方式从Hbase表中读取数据
 *         class DimAsyncFunction extends RichAsyncFunction implements DimJoinFunction{
 *             asyncInke{
 *                 CompleTableFuture
 *                 .supplyAsync         创建异步编排对象，有返回值
 *                 .thenApplyAsync      有入参，有返回值；上一个并行任务结果作为当前任务的入参
 *                 .thenAcceptAsync     有入参，没有返回值
 *             }
 *         }
 *         模板方法设计模式：在父类中定义完成某一个功能的核心算法的骨架(步骤)，但是某些步骤需要在子类中去实现，
 *                      在不改变父类核心算法骨架的前提下，每一个子类都可以有自己不同的实现。
 *
 *      将关联的结果写到Doris
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);

        //TODO 2.检查点相关的设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
       /* //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消后检查点是否保留
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        //TODO 3.从kafka主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_group";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //TODO 4.对流中数据类型进行转换并过滤掉空消息    jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
            new ProcessFunction<String, JSONObject>() {
                @Override
                public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                    if (StringUtils.isNotEmpty(jsonStr)) {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        out.collect(jsonObj);
                    }
                }
            }
        );
        // jsonObjDS.print(">>>");

        //TODO 5.按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS
            = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));

        //TODO 6.去重
        /*//6.1 状态 + 定时器   不足：每条数据到来后，都要5s后才会发送到下游，对时效性有影响
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastJsonObjState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                    lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject lastJsonObj = lastJsonObjState.value();
                    if (lastJsonObj == null) {
                        //说明没有重复，当前数据是第一条数据
                        //将当前这条数据放到状态中
                        lastJsonObjState.update(jsonObj);
                        //并注册一个5s后执行的定时器
                        long currentProcessingTime = ctx.timerService().currentProcessingTime();
                        ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                    } else {
                        //说明重复了（伪代码）
                        String ts1 = lastJsonObj.getString("聚合时间");
                        String ts2 = jsonObj.getString("聚合时间");
                        if (ts2.compareTo(ts1) >= 0) {
                            //如果第二条数据的聚合时间 > 状态中第一条数据的聚合时间，将第二条数据放到状态中
                            lastJsonObjState.update(jsonObj);
                        }
                    }
                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                    //定时器被触发执行的方法
                    //将状态中的数据发送到下游
                    JSONObject jsonObj = lastJsonObjState.value();
                    out.collect(jsonObj);
                    //清状态
                    lastJsonObjState.clear();
                }
            }
        );*/

        //6.2 状态 + 抵消
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastJsonObjState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                    lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {

                    JSONObject lastJsonObj = lastJsonObjState.value();
                    if (lastJsonObj != null) {
                        //重复了，将状态中影响到度量值的字段进行取反操作
                        String splitOriginalAmount = lastJsonObj.getString("split_original_amount");
                        String splitCouponAmount = lastJsonObj.getString("split_coupon_amount");
                        String splitActivityAmount = lastJsonObj.getString("split_activity_amount");
                        String splitTotalAmount = lastJsonObj.getString("split_total_amount");

                        lastJsonObj.put("split_original_amount", "-" + splitOriginalAmount);
                        lastJsonObj.put("split_coupon_amount", "-" + splitCouponAmount);
                        lastJsonObj.put("split_activity_amount", "-" + splitActivityAmount);
                        lastJsonObj.put("split_total_amount", "-" + splitTotalAmount);
                        //并将取反的数据发送到下游
                        out.collect(lastJsonObj);
                    }
                    out.collect(jsonObj);
                    lastJsonObjState.update(jsonObj);
                }
            }
        );
        // distinctDS.print(">>>>>");
        //TODO 7.再次对流中数据类型进行转换  jsonObj->实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> beanDS = distinctDS.map(
            new MapFunction<JSONObject, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                    // {"create_time":"2023-08-29 09:28:47","sku_num":"1","activity_rule_id":"1",
                    // "split_original_amount":"6499.0000","split_coupon_amount":"0.0","sku_id":"3",
                    // "date_id":"2023-08-29","user_id":"2728","province_id":"32","activity_id":"1",
                    // "sku_name":"小米1 ","id":"14256564","order_id":"54078","split_activity_amount":"500.0",
                    // "split_total_amount":"5999.0","ts":"1693531727"}
                    String skuId = jsonObj.getString("sku_id");
                    Long ts = jsonObj.getLong("ts") * 1000;
                    String splitOriginalAmount = jsonObj.getString("split_original_amount");
                    String splitCouponAmount = jsonObj.getString("split_coupon_amount");
                    String splitActivityAmount = jsonObj.getString("split_activity_amount");
                    String splitTotalAmount = jsonObj.getString("split_total_amount");
                    TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                        .skuId(skuId)
                        .originalAmount(new BigDecimal(splitOriginalAmount))
                        .couponAmount(new BigDecimal(splitCouponAmount))
                        .activityAmount(new BigDecimal(splitActivityAmount))
                        .orderAmount(new BigDecimal(splitTotalAmount))
                        .ts(ts)
                        .build();
                    return orderBean;
                }
            }
        );
        // beanDS.print(">>>>");
        //TODO 8.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<TradeSkuOrderBean>forMonotonousTimestamps()
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                        @Override
                        public long extractTimestamp(TradeSkuOrderBean orderBean, long recordTimestamp) {
                            return orderBean.getTs();
                        }
                    }
                )
        );
        //TODO 9.按照统计的维度sku进行分组
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS
            = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);

        //TODO 10.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS
            = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));


        //TODO 11.聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                    value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                    value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                    value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                    value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                    return value1;
                }
            },
            new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                @Override
                public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                    String stt = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());
                    String curDate = DateFormatUtil.toDate(window.getStart());
                    for (TradeSkuOrderBean orderBean : input) {
                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setCurDate(curDate);
                        out.collect(orderBean);
                    }
                }
            }
        );
        // reduceDS.print(">>>>");

        //TODO 12.关联sku维度
        /*SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                private Connection conn;
                private Jedis jedis;

                @Override
                public void open(Configuration parameters) throws Exception {
                    conn = HbaseUtil.getHbaseConnection();
                    jedis = RedisUtil.getJedis();
                }

                @Override
                public void close() throws Exception {
                    HbaseUtil.closeHbaseConnection(conn);
                    RedisUtil.closeJedis(jedis);
                }

                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                    //获取要关联的维度的主键
                    String skuId = orderBean.getSkuId();
                    //根据主键获取维度对象
                    JSONObject dimJsonObj = DimUtil.getDimInfo(jedis,conn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", skuId);
                    //将维度对象的属性补充到流中对象上
                    // id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                    return orderBean;
                }
            }
        );
        // withSkuInfoDS.print(">>>");
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new DimMapFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            }
        );

        // 将异步I/O操作应用于DataStream作为DataStream的一次转换操作
        AsyncDataStream.unorderedWait(
            reduceDS,
            //如何发送异步请求对流中数据进行处理  实现分发请求的 AsyncFunction
            new RichAsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                private StatefulRedisConnection<String,String> redisConn;
                private AsyncConnection hbaseConn;

                @Override
                public void open(Configuration parameters) throws Exception {
                    redisConn = RedisUtil.getAsyncRedisConnection();
                    hbaseConn = HbaseUtil.getAsyncConnection();
                }

                @Override
                public void close() throws Exception {
                    RedisUtil.closeAsyncRedisConn(redisConn);
                    HbaseUtil.closeAsyncConnection(hbaseConn);
                }

                @Override
                public void asyncInvoke(TradeSkuOrderBean orderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {
                    //创建CompletableFuture对象，有返回值
                    CompletableFuture.supplyAsync(
                        new Supplier<JSONObject>() {
                            @Override
                            public JSONObject get() {
                                //根据流中对象获取要关联的维度主键
                                String skuId = orderBean.getSkuId();
                                //根据维度的主键到Redis中获取对应的维度数据
                                JSONObject jsonObj = RedisUtil.asyncGetDimInfo(redisConn, "dim_sku_info:" + skuId);
                                return jsonObj;
                            }
                        }
                        //有入参，有返回值，将上一个并行任务的结果，作为参数传递到当前并行任务中
                    ).thenApplyAsync(
                        new Function<JSONObject, JSONObject>() {
                            @Override
                            public JSONObject apply(JSONObject jsonObj) {
                                //如果在Redis中没有获取到维度数据  发送请求到Hbase中查询维度
                                if(jsonObj == null){
                                    System.out.println("~~从hbase中获取维度数据~~");
                                    jsonObj = HbaseUtil.getDimInfoFromHbaseByAsync(hbaseConn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", orderBean.getSkuId());
                                    //并将查询结果放到Redis中缓存
                                    RedisUtil.asyncWriteDim(redisConn,"dim_sku_info:" + orderBean.getSkuId(),jsonObj);
                                }else{
                                    System.out.println("~~从Redis中获取维度数据~~");
                                }
                                return jsonObj;
                            }
                        }
                        //有入参，没有返回值
                    ).thenAcceptAsync(
                        new Consumer<JSONObject>() {
                            @Override
                            public void accept(JSONObject dimJsonObj) {
                                if(dimJsonObj != null){
                                    //将维度对象的属性补充到流中对象
                                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                                    //获取数据库交互的结果并发送给 ResultFuture的回调函数
                                    resultFuture.complete(Collections.singleton(orderBean));
                                }

                            }
                        }
                    );
                }
            },
            60,
            TimeUnit.SECONDS
        );
        */
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            },
            60, TimeUnit.SECONDS
        );
        // withSkuInfoDS.print(">>>>");
        //TODO 13.关联spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
            withSkuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setSpuName(dimInfoJsonObj.getString("spu_name"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSpuId();
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //TODO 14.关联tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
            withSpuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setTrademarkName(dimInfoJsonObj.getString("tm_name"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getTrademarkId();
                }
            },
            60,
            TimeUnit.SECONDS
        );
        //TODO 15.关联category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream.unorderedWait(
            withTmDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory3Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory3Name(dim.getString("name"));
                    bean.setCategory2Id(dim.getString("category2_id"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //TODO 16.关联category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream.unorderedWait(
            c3Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory2Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory2Name(dim.getString("name"));
                    bean.setCategory1Id(dim.getString("category1_id"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //TODO 17.关联category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> restulStream = AsyncDataStream.unorderedWait(
            c2Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory1Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory1Name(dim.getString("name"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //TODO 18.将结果写到Doris中
        restulStream.print(">>>");
        restulStream
            .map(new BeanToJsonStrFunction<>())
            .sinkTo(DorisUtil.getDorisSink("dws_trade_sku_order_window"));
        env.execute();
    }
}
